package org.diy.traffic.daemon;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Level;

import org.diy.traffic.config.DataBotConfig;
import org.diy.traffic.config.DataBotConfig.TrafficConfigException;
import org.diy.traffic.sinks.TrafficSink;
import org.diy.traffic.sources.TrafficSource;
import org.diy.traffic.updates.TrafficUpdate;
import org.diy.traffic.util.TLog;

/**
 * This class keeps track of the connections between inputs (sources) and outputs (sinks).
 * 
 * It has several methods which are called by client code to register new sources and sinks, and establish
 * connections between them.
 * @author Dan Greenblatt
 *
 */

public class AirTrafficController {
	
	private static AirTrafficController myController = null;
	
	/** Default value (in milliseconds) for the interval on which the input daemon checks TrafficSources for updates **/
	private static final long defaultInputDInterval =  10000;
	
	/** Default value (in milliseconds) for the interval on which the output daemon passes updates to TrafficSinks **/
	private static final long defaultOutputDInterval = 10000;
	
	/** Keeps track of all the registered traffic sources. Maps a unique name (string) to the instance **/
	private HashMap<String, TrafficSource> trafficSources   = new HashMap<String, TrafficSource>();
	
	/** Keeps track of all the registered traffic sinks. Maps a unique name (string) to the instance **/
	private HashMap<String, TrafficSink>   trafficSinks     = new HashMap<String,TrafficSink>();

	/** Thread-safe queue which takes updates from sources and feeds them to the sinks **/
	private ArrayBlockingQueue<TrafficUpdate> updateQueue = new ArrayBlockingQueue<TrafficUpdate>(1000);
		
	/** Maintains the mapping of inputs to outputs, as designated by client code **/
	private HashMap<TrafficSource, ArrayList<TrafficSink>> sourceSinkMap = 
		new HashMap<TrafficSource, ArrayList<TrafficSink>>();
	
	
	/** Thread polls the sources for new updates and puts them in the updateQueue **/
	private InputDaemon  inputD; 
	private long inputInterval = AirTrafficController.defaultInputDInterval;
	private long outputInterval = AirTrafficController.defaultOutputDInterval;
	
	/** Thread takes updates from the updateQueue and hands them off to the appropriate sinks **/
	private OutputDaemon outputD;
	
	/**
	 * Constructor for the AirTrafficController uses default values for input and output daemon intervals
	 */
	public AirTrafficController()  {
		this(defaultInputDInterval, defaultOutputDInterval);
	}
	
	/**
	 * Constructor for the AirTrafficController
	 * @param inputInterval  duration in milliseconds in between each round of checking traffic sources
	 * @param outputInterval duration in milliseconds in between each round of updating traffic sinks
	 */
	public AirTrafficController(long inputInterval, long outputInterval)  {
		/** put this off until .start() so the ATC has had a chance to be configurated **/
		//inputD  = new InputDaemon(inputInterval, this);
		//outputD = new OutputDaemon(outputInterval, this);
		
		this.inputInterval  = inputInterval;
		this.outputInterval = outputInterval;
	}
	
	public static AirTrafficController createController(long inputInterval, long outputInterval) {
		if (myController != null) {
			myController = null;
		} 
		myController = new AirTrafficController(inputInterval, outputInterval);
		return myController;
	}
	
	public static AirTrafficController getController() {
		return myController;
	}
	
	/**
	 * Clears the ATC of all it's state, including registered sources, registered sinks and mappings
	 */
	public void clear() {
		updateQueue.clear();
		trafficSinks.clear();
		trafficSources.clear();
		sourceSinkMap.clear();
	}
	
	/**
	 * This should be called by client code when all the inputs and outputs have been registered and mapped.
	 */
	public void start() {
		if (inputD == null)
			inputD  = new InputDaemon(inputInterval, this);
		inputD.start();

		if (outputD == null) 
			outputD = new OutputDaemon(outputInterval, this);
		outputD.start();
	}
	
	/*public void revive() {
		inputD.doRevive();
		inputD.notify();
		
		outputD.doRevive();
		outputD.notify();		
	}*/
	
	public void suspend() throws InterruptedException {
		TLog.log(Level.FINEST, "Suspending InputDaemon");
		inputD.doSuspend();
		TLog.log(Level.FINEST, "Interrupting InputDaemon");
		inputD.interrupt();
		TLog.log(Level.FINEST, "Waiting for InputDaemon to join");
		inputD.join();
		TLog.log(Level.FINEST, "InputDaemon joined");
		inputD = null;
		
		TLog.log(Level.FINEST, "Suspending OutputDaemon");
		outputD.doSuspend();
		TLog.log(Level.FINEST, "Interrupting OutputDaemon");
		outputD.interrupt();
		TLog.log(Level.FINEST, "Waiting for OutputDaemon to join");
		outputD.join();
		TLog.log(Level.FINEST, "OutputDaemon joined");
		outputD = null;
	}
	
	/*
	 * Should gracefully stop the ATC; not implemented
	 */
	public void stop() {}
	
	
	public HashMap<String, String> configureFromFile (String configFilePath) throws TrafficConfigException {
		DataBotConfig confX = new DataBotConfig(this);
		HashMap<String,String> botConfigParams = confX.configureFromLocalFile(configFilePath);	
	
		if (botConfigParams.get("inputInterval") != null) {
			this.inputInterval = Integer.parseInt(botConfigParams.get("inputInterval")) * 1000;
		}
		if (botConfigParams.get("outputInterval") != null) {
			this.outputInterval = Integer.parseInt(botConfigParams.get("outputInterval")) * 1000;
		}
		return botConfigParams;
	}
	
	
	/*
	 * Returns a list containing references of all the registered traffic sources 	
	 */
	public ArrayList<TrafficSource> getTrafficSources() {
		return new ArrayList<TrafficSource>(this.trafficSources.values());
	}
	
	/*
	 * Returns a list containing references of all the registered traffic sinks
	 */
	public ArrayList<TrafficSink> getTrafficSinks() {
		return new ArrayList<TrafficSink>(this.trafficSinks.values());
	}
	
	/*
	 * Called by a source to add an update to the ATC's updateQueue
	 * 
	 * @return true if the update was added, false if not
	 */
	public boolean addUpdate(TrafficUpdate update) {
		return this.updateQueue.add(update);
	}
	
	/*
	 * Called by a source to add a list of updates to the ATC's updateQueue
	 * 
	 * @return true if all the updates were added, false if not
	 */
	public boolean addUpdates(List<TrafficUpdate> updates) {
		return this.updateQueue.addAll(updates);
	}
	
	/*
	 * Removes all the updates currently in the updateQueue and returns them.
	 * 
	 * @return a list of all the updates which were in the update queue. If there were no updates, the list is empty
	 */
	public List<TrafficUpdate> getAllUpdates() {
		List<TrafficUpdate> l = new ArrayList<TrafficUpdate>();
		this.updateQueue.drainTo(l);
		return l;
	}
	
	/*
	 * Called by a sink to get a single update from the queue. Note that the update is removed from the queue.
	 * 
	 * @return a single traffic update from the queue, or null if there is no update
	 */
	public TrafficUpdate getSingleUpdate() {
		TrafficUpdate t = this.updateQueue.poll();
		return t;
	}
	
	
	/** This section contains methods for de-/registering Broadcast Traffic Sources **/
	
	/*
	 * Register a traffic source with the ATC. Note, traffic sources must have a unique name as returned by TrafficSource.getName()
	 * @param t the traffic source to register
	 * @throws an ATCException if a source with the same name is already registered
	 */
	public void registerTrafficSource(TrafficSource t) throws ATCException{
		if (isTrafficSourceRegistered(t)) { 
			throw new ATCException("Traffic source " + t + " is already registered");
		} else {
			this.trafficSources.put(t.getName(), t);
		}
	}
	
	/*
	 * Deregister a traffic source from the ATC
	 * @param t The TrafficSource to de-register
	 * @throws An ATCException if the traffic source is not actually registered
	 */
	public void deregisterTrafficSource(TrafficSource t) throws ATCException {
		if (!trafficSources.containsKey(t.getName()))  {
			throw new ATCException("Can't deregister traffic source " + t + "; not registered");
		} else {
			trafficSources.remove(t.getName());
		}
	}
	
	/*
	 * Checks to see if the traffic source is registered, keyed on the value returned by TrafficSource.getName()
	 * @return true if the source is registered, false if not
	 */
	public boolean isTrafficSourceRegistered(TrafficSource t){
		return trafficSources.containsKey(t.getName());
	}
	
	
	/** This section contains methods for de-/registering Broadcast Traffic Sinks **/
	
	/*
	 * Register a traffic sink with the ATC
	 * @param t The TrafficSink to register
	 * @throws ATCException if a TrafficSink with the same name (as returned by TrafficSink.getName()) is already registered
	 */
	public void registerTrafficSink(TrafficSink t) throws ATCException{
		if (isTrafficSinkRegistered(t)) { 
			throw new ATCException("Traffic sink " + t + " is already registered");
		} else {
			this.trafficSinks.put(t.getName(), t);
		}
	}
	
	/*
	 * De-register a traffic sink from the ATC
	 * @param t The TrafficSink to de-register
	 * @throws ATCException if a TrafficSink with the same name as t is not already registered
	 */
	public void deregisterBroadcastSink(TrafficSink t) throws ATCException {
		if (!trafficSinks.containsKey(t.getName()))  {
			throw new ATCException("Can't deregister traffic sink" + t + "; not registered");
		} else {
			trafficSinks.remove(t.getName());
		}
	}
	
	/*
	 * Checks to see if the traffic sink is registered, keyed on the value returned by TrafficSink.getName()
	 * @return true if the sink is registered, false if not
	 */
	public boolean isTrafficSinkRegistered(TrafficSink t) {
		return trafficSinks.containsKey(t.getName());
	}
	
	
	/*
	 * Returns a list of TrafficSinks registered for a given source
	 * @param source The TrafficSource for which you want to find a list of registered sinks
	 * @return an ArrayList of TrafficSinks, or null if no sinks are registered for the source
	 */
	public ArrayList<TrafficSink> getSinksForSource(TrafficSource source) {
		return sourceSinkMap.get(source);
	}

	
	/*
	 * Called by client code to establish a relationship between a sink and a source
	 * @param tSource The TrafficSource where updates will be coming from
	 * @param tSink   The TrafficSink where updates will be going to
	 * @throws ATCException if the sink is already registered for the source
	 */
	public void establishPipe(TrafficSource tSource, TrafficSink tSink) throws ATCException {
		if (sourceSinkMap.containsKey(tSource)) {
			ArrayList<TrafficSink> mappedSinks = sourceSinkMap.get(tSource);
			
			if (mappedSinks.contains(tSink)) {			
				throw new ATCException("Can't establish rel'p between " + 
						tSource.getName() + " and " + tSink.getName() +  "; source already registered");
			} else {
				mappedSinks.add(tSink);
			}
		} else {
			ArrayList<TrafficSink> sinkList = new ArrayList<TrafficSink>();
			sinkList.add(tSink);
			sourceSinkMap.put(tSource, sinkList);
		}
	}
	
	/*
	 * Called by client code to break a relationship between a sink and a source
	 * @param tSource The source from which tSink should be removed as a target for TrafficUpdates
	 * @param tSink The sink which should be removed as a target of TrafficUpdates from tSource
	 * @throws ATCException if tSource isn't registered at all, or tSink isn't specifically registered as a
	 * 		sink for tSource
	 */
	public void breakPipe(TrafficSource tSource, TrafficSink tSink) throws ATCException {
		if (!sourceSinkMap.containsKey(tSource)) {
			throw new ATCException("Can't break rel'p between " + 
					tSource.getName() + " and " + tSink.getName() +  "; source not registered");
		} else {
			ArrayList<TrafficSink> registeredSinks = sourceSinkMap.get(tSource);
			if (!registeredSinks.contains(tSink)) {
				throw new ATCException("Can't break rel'p between " + 
						tSource.getName() + " and " + tSink.getName() +  "; sink not registered for source");
			} else {
				registeredSinks.remove(tSink);
			}
		}
	}
	
	/*
	 * An Exception used to represents error states which occur in the AirTrafficController class
	 */
	public class ATCException extends Exception {
		public ATCException(String msg) {
			super(msg);
		}
	}
}

